/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.analysys.presto.connector.hbase.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.analysys.presto.connector.hbase.HBaseColumnMetadata;
import com.analysys.presto.connector.hbase.meta.TableMetaInfo;
import com.analysys.presto.connector.hbase.ConditionInfo;
import com.facebook.presto.spi.ColumnMetadata;
import com.facebook.presto.common.type.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import io.airlift.log.Logger;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;

import javax.validation.constraints.NotNull;
import java.io.File;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import static com.analysys.presto.connector.hbase.utils.Constant.*;

/**
 * utils
 *
 * @author wupeng
 * @date 2019/01/29
 */
public class Utils {

    public static final Logger logger = Logger.get(Utils.class);

    /**
     * Read table json from metaDir by schema name and table name
     *
     * @param schemaName schema name
     * @param tableName  table name
     * @param metaDir    meta dir
     * @return json file content
     */
    private static JSONContent readTableJson(String schemaName, String tableName, String metaDir) {
        try {
            String tableMetaDir = metaDir + File.separator
                    + (StringUtils.isBlank(schemaName) ? DEFAULT_HBASE_NAMESPACE_NAME : schemaName);
            File metaFile = new File(tableMetaDir, tableName + TABLE_META_FILE_TAIL);
            if(metaFile.exists()) {
                final JSONContent jsonContent = new JSONContent(FileUtils.readFileToString(metaFile, JSON_ENCODING_UTF8), metaFile.length(), metaFile.lastModified());
                jsonContent.setFileName(metaFile.getName());
                return jsonContent;
            }

            final List<String> files = listTableJson(schemaName, metaDir);
            for (String file : files) {
                String fileName = file.replaceAll(TABLE_META_FILE_TAIL, "");
                if(tableName.startsWith(fileName)) {
                    metaFile = new File(tableMetaDir, file);
                    final JSONContent jsonContent = new JSONContent(FileUtils.readFileToString(metaFile, JSON_ENCODING_UTF8), metaFile.length(), metaFile.lastModified());
                    jsonContent.setFileName(metaFile.getName());
                    return jsonContent;
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return new JSONContent("", 0, 0);
    }


    /**
     * Read table json from metaDir by schema name and table name
     *
     * @param schemaName schema name
     * @param metaDir    meta dir
     * @return json file content
     */
    @NotNull
    private static List<String> listTableJson(String schemaName, String metaDir) {
        try {
            String tableMetaPath = metaDir + File.separator
                    + (StringUtils.isBlank(schemaName) ? DEFAULT_HBASE_NAMESPACE_NAME : schemaName);
            File[] files = new File(tableMetaPath).listFiles((file) -> {
                return file.isFile() && !file.getName().startsWith(".") && file.getName().endsWith(TABLE_META_FILE_TAIL);
            });
            if(files != null) {
                return Arrays.stream(files).map(File::getName).collect(Collectors.toList());
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return Collections.emptyList();
    }

    /**
     * Read table json from metaDir by schema name and table name.
     * And convert it to an Object of TableMetaInfo.
     *
     * @param schemaName schema name
     * @param tableName  table name
     * @param metaDir    meta info dir
     * @return Object of TableMetaInfo
     */
    public static TableMetaInfo getTableMetaInfoFromJson(String schemaName, String tableName, String metaDir) {
        long startTime = System.currentTimeMillis();
        try {
            ObjectMapper mapper = new ObjectMapper();
            final JSONContent jsonContent = readTableJson(schemaName, tableName, metaDir);
            String jsonString = jsonContent.getContent();
            final TableMetaInfo tableMetaInfo = mapper.readValue(jsonString, TableMetaInfo.class);
            tableMetaInfo.setTableName(tableName);
            return tableMetaInfo;
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return null;
    }


    /**
     * Read table json from metaDir by schema name and table name.
     * And convert it to an Object of TableMetaInfo.
     *
     * @param schemaName schema name
     * @param metaDir    meta info dir
     * @return Object of TableMetaInfo
     */
    public static Map<String, TableMetaInfo> listTableMetaInfoFromJson(String schemaName, String metaDir) {
        schemaName = StringUtils.isBlank(schemaName) ? DEFAULT_HBASE_NAMESPACE_NAME : schemaName;
        Map<String, TableMetaInfo> tableMap = new HashMap<>();
        try {
            ObjectMapper mapper = new ObjectMapper();
            final List<String> tableJsons = listTableJson(schemaName, metaDir);
            for (String tableName : tableJsons) {
                final JSONContent jsonContent = readTableJson(schemaName, tableName.replaceAll(TABLE_META_FILE_TAIL, ""), metaDir);
                String jsonString = jsonContent.getContent();
                final TableMetaInfo tableMetaInfo = mapper.readValue(jsonString, TableMetaInfo.class);
                tableMetaInfo.setLastModify(jsonContent.getLastModify());
                tableMap.put(schemaName+":"+tableMetaInfo.getTableName(), tableMetaInfo);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return tableMap;
    }

    /**
     * Fetch column meta info from json file
     *
     * @param schemaName schema name
     * @param tableName  table name
     * @param metaDir    meta dir
     * @return list of column meta info
     */
    public static ImmutableList<ColumnMetadata> getColumnMetaFromJson(String schemaName, String tableName, String metaDir) {
        long startTime = System.currentTimeMillis();
        ImmutableList.Builder<ColumnMetadata> columnsMetadata = ImmutableList.builder();
        try {
            String metaFileName = tableName;

            // 从 JSON 文件中获取 meta 信息
            final Map<String, TableMetaInfo> metaInfoMap = Utils.listTableMetaInfoFromJson(schemaName, metaDir);
            // 从 Hbase meta 中获取的表信息
            String schemaAndName = schemaName + ":" + tableName;
            for (Map.Entry<String, TableMetaInfo> entry : metaInfoMap.entrySet()) {
                if(schemaAndName.startsWith(entry.getKey())) {
                    metaFileName = entry.getValue().getTableName();
                    break;
                }
            }
            final JSONContent jsonContent = readTableJson(schemaName, metaFileName, metaDir);
            String jsonStr = jsonContent.getContent();
            JSONObject obj = (JSONObject) JSON.parse(jsonStr);
            JSONArray cols = obj.getJSONArray(JSON_TABLEMETA_COLUMNES);
            boolean hasRowKey = false;
            for (int i = 0; i < cols.size(); i++) {
                JSONObject temp = (JSONObject)cols.get(i);
                String family = temp.getString(JSON_TABLEMETA_FAMILY);
                String columnName = temp.getString(JSON_TABLEMETA_COLUMNNAME);
                String type = temp.getString(JSON_TABLEMETA_TYPE);
                boolean isRowKey = temp.getBoolean(JSON_TABLEMETA_ISROWKEY);
                final HBaseColumnMetadata columnMetadata = new HBaseColumnMetadata(family, columnName, matchType(type), isRowKey);
                columnsMetadata.add(columnMetadata);
                if (isRowKey)  {
                    hasRowKey = true;
                }
            }
            Preconditions.checkState(hasRowKey,
                    "Table %s.%s doesn't specified ROW_KEY column." +
                            " Specify ROW_KEY in your .json file.", schemaName, tableName);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
        return columnsMetadata.build();
    }

    /**
     * Add zero prefix to a salt string
     *
     * @param key       key
     * @param missCount missCount
     * @return standard salt value
     */
    public static String addZeroPrefix(String key, int missCount) {
        for (int j = missCount; j > 0; j--) {
            key = "0" + key;
        }
        return key;
    }

    /**
     * Find the presto type of column you configured in json file by type flag.
     *
     * @param type The type value that configured in json file.
     * @return type in presto
     */
    public static com.facebook.presto.common.type.Type matchType(String type) {
        if (type == null) {
            return VarcharType.VARCHAR;
        }

        switch (type.toLowerCase()) {
            case "string":
                return VarcharType.VARCHAR;
            case "int":
            case "integer":
                return IntegerType.INTEGER;
            case "bigint":
                return BigintType.BIGINT;
            case "double":
            case "float":
            case "numeric":
                return DoubleType.DOUBLE;
            case "boolean":
                return BooleanType.BOOLEAN;
            case "array<string>":
            case "array":
                return new ArrayType(VarcharType.VARCHAR);
            case "timestamp":
            case "datetime":
                return TimestampType.TIMESTAMP;
            case "date":
                return DateType.DATE;
            case "number":
                return DecimalType.createDecimalType(DECIMAL_DEFAULT_PRECISION, DECIMAL_DEFAULT_SCALE);
            default:
                return VarcharType.VARCHAR;
        }
    }

    /**
     * Whether sql constraint contains conditions like "rowKey='xxx'" or "rowKey in ('xxx','xxx')"
     *
     * @return true if this sql is batch get.
     */
    public static boolean isBatchGet(List<ConditionInfo> conditions, String rowKeyColName) {
        if (conditions != null) {
            for (ConditionInfo cond : conditions) {
                if (rowKeyColName.equalsIgnoreCase(cond.getColName())
                        && cond.getOperator() == CONDITION_OPER.EQ) {
                    return true;
                }
            }
        }
        return false;
    }

    /**
     * Copy contents in ${srcAry} from position ${srcPos} for ${length} bytes.
     *
     * @param srcAry source array
     * @param srcPos start position
     * @param length length
     * @return copied byte array
     */
    public static byte[] arrayCopy(byte[] srcAry, int srcPos, int length) {
        byte[] destAry = new byte[length];
        System.arraycopy(srcAry, srcPos, destAry, 0, length);
        return destAry;
    }

    /**
     * 去掉array<string>中\001分隔的元素前的一个空格
     * presto查询出来的值是"aaa\001 bbb\001 ccc"，元素之间会带上一个空格
     *
     * @param value value
     * @return string
     */
    public static String removeExtraSpaceInArrayString(String value) {
        StringBuilder buff = new StringBuilder();
        String[] ss = value.split(ARRAY_STRING_SPLITTER);
        for (int j = 0; j < ss.length; j++) {
            String ele = ss[j];
            if (j > 0) {
                ele = ele.substring(1, ele.length());
            } else {
                buff.append(ARRAY_STRING_SPLITTER);
            }
            buff.append(ele);
        }
        return buff.toString();
    }

}



